package net.chengp.demo;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.NamespaceExistException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

public class HBaseUtils {

	private static Configuration conf;
	private static Connection connection;
	private static Admin admin;

	private static boolean enableKerberos = true;

	static {
		conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "cdh1");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		// 开启kerberos认证
		if (enableKerberos) {
			conf.addResource(new Path("D:/cdh_kerberos/139/hbase-site.xml"));
			conf.set("username.client.keytab.file", "D:/cdh_kerberos/139/hbase.keytab");
			conf.set("username.client.kerberos.principal", "hbase/cdh2@LEGION.COM");
			conf.set("hadoop.security.authentication", "Kerberos");
			conf.set("hbase.security.authentication", "Kerberos");
			System.setProperty("java.security.krb5.conf", "D:/cdh_kerberos/139/krb5.conf");
			UserGroupInformation.setConfiguration(conf);
			try {
				UserGroupInformation.loginUserFromKeytab("hbase/cdh2@LEGION.COM",
						"D:/cdh_kerberos/139/hbase.keytab");
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		// 获取连接
		try {
			connection = ConnectionFactory.createConnection(conf);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 获取HBase管理员
	 * 
	 * @return
	 * @throws IOException
	 */
	public static Admin getAdmin() throws IOException {
		return connection.getAdmin();
	}

	/**
	 * 创建命名空间
	 * 
	 * @param spaceName
	 */
	public static void createNameSpace(String spaceName) {
		NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(spaceName).build();
		try {
			admin.createNamespace(namespaceDescriptor);
			System.out.println("创建命名空间" + spaceName + "成功！");
		} catch (NamespaceExistException ex) {
			System.out.println("命名空间已存在！");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 创建表
	 * 
	 * @param tableName
	 * @param cfs
	 */
	public static void createTable(String tableName, String... cfs) {
		if (cfs == null || cfs.length == 0) {
			System.out.println("列族不能为空！");
			return;
		}
		TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
		List<ColumnFamilyDescriptor> cFDBList = new ArrayList<>();
		for (String cf : cfs) {
			cFDBList.add(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf)).setMaxVersions(3).build());
		}
		tableDescriptorBuilder.setColumnFamilies(cFDBList);
		try {
			if (isTableExist(tableName)) {
				System.out.println("表已经存在！");
				admin.modifyTable(tableDescriptorBuilder.build());
			}else {
				System.out.println("表不存在！");
				admin.createTable(tableDescriptorBuilder.build());
			}
			System.out.println("创建表" + tableName + "成功！");
		} catch (IOException e) {
			System.out.println("创建失败！");
			e.printStackTrace();
		}
	}

	/**
	 * 判断表是否存在
	 * 
	 * @param tableName
	 * @return
	 */
	public static boolean isTableExist(String tableName) {
		try {
			return admin.tableExists(TableName.valueOf(tableName));
		} catch (IOException e) {
			e.printStackTrace();
		}
		return false;
	}

	/**
	 * 列出所有表
	 */
	public static void listTables() {
		try {
			TableName[] tableNames = admin.listTableNames();
			System.out.println("打印所有命名空间表名：");
			for (TableName tableName : tableNames) {
				System.out.println(tableName);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 列出指定命名空间里的表
	 * 
	 * @param spaceName
	 */
	public static void listTablesByNameSpace(String spaceName) {
		if (spaceName == null || spaceName.length() == 0) {
			System.out.println("请输入正确的命名空间！");
			return;
		}
		try {
			TableName[] tableNames = admin.listTableNamesByNamespace(spaceName);
			System.out.println("打印" + spaceName + "命名空间下表名：");
			for (TableName tableName : tableNames) {
				System.out.println(tableName);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 删除指定row
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param colFamily
	 * @param cols
	 */
	public static void deleteRow(String tableName, String rowKey, String colFamily, String... cols) {
		if (tableName == null || tableName.length() == 0) {
			System.out.println("请正确输入表名！");
			return;
		}
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Delete del = new Delete(Bytes.toBytes(rowKey));
			// 删除指定列族
			del.addFamily(Bytes.toBytes(colFamily));
			// 删除指定列
			for (String col : cols) {
				del.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col));
			}
			table.delete(del);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			closeTable(table);
		}
	}

	/**
	 * 扫描表
	 * 
	 * @param tableName
	 * @param startRow
	 *            起始位置
	 * @param stopRow
	 *            结束位置
	 */
	public static void scanTable(String tableName, String family, String startRow, String stopRow, int limit) {
		if (tableName == null || tableName.length() == 0) {
			System.out.println("请正确输入表名！");
			return;
		}
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			// 左闭右开
			if (startRow != null && stopRow != null) {
				scan.withStartRow(Bytes.toBytes(startRow));
				scan.withStopRow(Bytes.toBytes(stopRow));
			} else {
				startRow = "无效";
				stopRow = "无效";
			}
			scan.setLimit(limit);
			scan.addFamily(Bytes.toBytes(family));
			ResultScanner resultScanner = table.getScanner(scan);
			Iterator<Result> iterator = resultScanner.iterator();
			System.out.println("scan\t startRow: " + startRow + "\t stopRow: " + stopRow);
			System.out.println("RowKey\tTimeStamp\tcolumnFamilyName\tcolumnQualifierName");
			while (iterator.hasNext()) {
				Result result = iterator.next();
				showCell(result);
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			closeTable(table);
		}
	}

	/**
	 * 格式化输出
	 * 
	 * @param result
	 */
	public static void showCell(Result result) {
		Cell[] cells = result.rawCells();
		for (Cell cell : cells) {
			System.out.print(Bytes.toString(CellUtil.cloneRow(cell)) + "\t");
			System.out.print(cell.getTimestamp() + "\t");
			String columnFamilyName = Bytes.toString(CellUtil.cloneFamily(cell));
			String columnQualifierName = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));
			System.out.println(columnFamilyName + ":" + columnQualifierName + "\t\t\t" + value);
		}
	}

	/**
	 * 关闭连接
	 */
	public static void closeConn() {
		try {
			if (null != admin) {
				admin.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}

	}

	/**
	 * 关闭表
	 * 
	 * @param table
	 */
	public static void closeTable(Table table) {
		if (table != null) {
			try {
				table.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 插入信息
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param cf
	 * @param column
	 * @param value
	 */
	public static void putTableData(String tableName, String rowKey, String cf, String column, String value) {
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Put p = new Put(Bytes.toBytes(rowKey));
			p.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(value));
			table.put(p);
			System.out.println("插入数据成功！");
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			closeTable(table);
		}
	}

	/**
	 * 获得指定row
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param colFamily
	 * @param col
	 */
	public static void getRow(String tableName, String rowKey, String colFamily, String col) {
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Get g = new Get(Bytes.toBytes(rowKey));
			g.setMaxVersions(3);
			// 获取指定列族数据
			if (col == null && colFamily != null) {
				g.addFamily(Bytes.toBytes(colFamily));
			} else if (col != null && colFamily != null) {
				// 获取指定列数据
				g.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col));
			}
			Result result = table.get(g);
			System.out.println("getRow\t");
			System.out.println("RowKey\tTimeStamp\tcolumnFamilyName\tcolumnQualifierName");
			showCell(result);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			closeTable(table);
		}
	}

	/**
	 * 删除表
	 * 
	 * @param tableName
	 */
	public static void deleteTable(String tableName) {
		try {
			if (!isTableExist(tableName)) {
				return;
			}
			admin.disableTable(TableName.valueOf(tableName));
			admin.deleteTable(TableName.valueOf(tableName));

		} catch (IOException e) {
			e.printStackTrace();
		} finally {

		}
	}

	public static void main(String[] args) {
		try {
			admin = getAdmin();
			TableName[] tables = admin.listTableNames();
			System.out.println(Arrays.toString(tables));
			// System.out.println(admin.getDescriptor(tables[2]));
			System.out.println("--------------------------------------------------");
			createNameSpace("homework");
			System.out.println("--------------------------------------------------");
			createTable("test", "info", "aaa");
			createTable("homework:t1", "f1", "f2");
			createTable("tab1", "f1");
			createTable("homework:stu", "info");
			System.out.println("--------------------------------------------------");
			listTables();
			System.out.println("--------------------------------------------------");
			listTablesByNameSpace("homework");
			System.out.println("--------------------------------------------------");
			putTableData("homework:stu", "10001", "info", "name", "tom");
			putTableData("homework:stu", "10001", "info", "age", "21");
			System.out.println("--------------------------------------------------");
			putTableData("homework:stu", "10001", "info", "math", "101");
			System.out.println("--------------------------------------------------");
			putTableData("homework:stu", "10001", "info", "age", "22");
			System.out.println("--------------------------------------------------");
			putTableData("homework:t1", "10001", "f1", "name", "张三");
			putTableData("homework:t1", "10001", "f2", "name", "李四");
			putTableData("homework:t1", "10001", "f1", "age", "22");
			putTableData("homework:t1", "10001", "f2", "name", "李四四");
			
			putTableData("tab1", "10001", "f1", "data", "abc");
			putTableData("tab1", "10001", "f1", "data", "abcd");
			scanTable("homework:stu", "info", null, null, 10);
			System.out.println("--------------------------------------------------");
			getRow("homework:stu", "10001", "info", "age");
			getRow("homework:t1", "10001", "f2", "name");
			getRow("tab1", "10001", "f1", "data");
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			closeConn();
		}
	}
}
